package com.tca.common.learning.hadoop.hdfs;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;

import java.io.IOException;
import java.net.URI;

/**
 * @author zhoua
 * @date 2023/10/28 21:52
 */
@Slf4j
public class HdfsClient {

    private final FileSystem fileSystem;

    public HdfsClient(URI uri, String username) throws IOException, InterruptedException {
        this.fileSystem = FileSystem.get(uri, new Configuration(), username);
    }

    /**
     * 创建路径
     * @param path
     */
    public void mkdirs(String path)  {

        try {
            fileSystem.mkdirs(new Path(path));
        } catch (Exception e) {
            log.error("操作失败", e);
        } finally {
            try {
                fileSystem.close();
            } catch (IOException e) {
                log.error("关闭 filesystem 失败", e);
            }
        }

    }

    /**
     * 本地文件上传
     * @param localFile
     * @param destFile
     */
    public void copyFromLocalFile(String localFile, String destFile) {
        try {
            fileSystem.copyFromLocalFile(new Path(localFile), new Path(destFile));
        } catch (Exception e) {
            log.error("操作失败", e);
        } finally {
            try {
                fileSystem.close();
            } catch (IOException e) {
                log.error("关闭 filesystem 失败", e);
            }
        }

    }

    /**
     * 下载到本地*
     * @param destPath
     * @param localPath
     */
    public void copy2LocalFile(String destPath, String localPath) {
        try {
            fileSystem.copyToLocalFile(
                    false,
                    new Path(destPath),
                    new Path(localPath),
                    true
            );
        } catch (Exception e) {
            log.error("操作失败", e);
        } finally {
            try {
                fileSystem.close();
            } catch (IOException e) {
                log.error("关闭 filesystem 失败", e);
            }
        }
    }

    /**
     * rename*
     * @param srcName
     * @param destName
     */
    public void rename(String srcName, String destName) {
        try {
            fileSystem.rename(new Path(srcName), new Path(destName));
        } catch (Exception e) {
            log.error("操作失败", e);
        } finally {
            try {
                fileSystem.close();
            } catch (IOException e) {
                log.error("关闭 filesystem 失败", e);
            }
        }

    }

    /**
     * 删除文件或目录(递归删除)
     * @param path
     */
    public void delete(String path) {
        try {
            fileSystem.delete(new Path(path), true);
        } catch (Exception e) {
            log.error("操作失败", e);
        } finally {
            try {
                fileSystem.close();
            } catch (IOException e) {
                log.error("关闭 filesystem 失败", e);
            }
        }

    }

    /**
     * listFiles
     * @param path
     */
    public void listFiles(String path) {
        try {
            RemoteIterator<LocatedFileStatus> iterator = fileSystem.listFiles(new Path(path), true);
            while (iterator.hasNext()) {
                LocatedFileStatus locatedFileStatus = iterator.next();
                if (locatedFileStatus.isFile()) {
                    log.info(locatedFileStatus.getPath().getName());
                } else if (locatedFileStatus.isDirectory()){
                    listFiles(locatedFileStatus.getPath().toString());
                }
            }
        } catch (Exception e) {
            log.error("操作失败", e);
        } finally {
            try {
                fileSystem.close();
            } catch (IOException e) {
                log.error("关闭 filesystem 失败", e);
            }
        }

    }
}
